package com.shujia.mllib

import org.apache.spark.ml.classification.LogisticRegressionModel
import org.apache.spark.ml.linalg
import org.apache.spark.ml.linalg.Vectors
import org.apache.spark.sql.{DataFrame, Row, SparkSession}

object Demo6ImageModelPredict {
  def main(args: Array[String]): Unit = {


    val spark: SparkSession = SparkSession
      .builder()
      .master("local[*]")
      .appName("Demo4ReadImage")
      .config("spark.sql.shuffle.partitions", 4)
      .getOrCreate()

    /**
     * 1、加载模型
     */
    val model: LogisticRegressionModel = LogisticRegressionModel.load("spark/data/imageModel")


    /**
     * 2、将需要预测的数据加载进来，并进行同样的特性工程处理
     */
    val image: DataFrame = spark
      .read
      .format("image")
      .load("spark/data/42.jpg")

    import spark.implicits._
    import org.apache.spark.sql.functions._

    val imageDF: DataFrame = image
      .select($"image.origin", $"image.data")
      .as[(String, Array[Byte])]
      .map {
        case (filePath: String, bytes: Array[Byte]) => {
          val ints: Array[Int] = bytes.map(_.toInt)
          val doubles: Array[Double] = ints.map(i => {
            if (i >= 0) {
              0.0
            } else {
              1.0
            }
          })
          // 从文件的绝对路径中提取文件名 用于关联 提取label
          val filename: String = filePath.split("/").last
          // 将处理好的图片数据转换成 稠密向量
          // 转成稀疏向量 再以libsvm格式保存 重新读取的时候会造成长度的丢失
          val image_vec: linalg.Vector = Vectors.dense(doubles)
          (filename, image_vec)
        }
      }.toDF("filename", "features")


    /**
     * 3、使用模型进行预测
     */
    val resDF: DataFrame = model.transform(imageDF)
    resDF.show()

  }

}
